package com.lxw.multidb.modules.es;


import cn.tannn.jdevelops.exception.built.BusinessException;
import cn.tannn.jdevelops.result.utils.UUIDUtils;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Refresh;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.IndexSettings;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.lxw.multidb.controller.es.CreateIndexDTO;
import com.lxw.multidb.controller.es.ImportProgressDTO;
import com.lxw.multidb.modules.es.util.EsMappingUtil;
import com.lxw.multidb.modules.es.util.EsStrUtil;
import com.lxw.multidb.modules.file.DynFileOperateService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * 数据ES索引操作Service
 * 提供Elasticsearch索引的创建、数据导入等核心功能
 * <p>
 * 主要功能：
 * 1. 从配置文件加载索引映射规则
 * 2. 创建Elasticsearch索引结构
 * 3. 从关系型数据库批量导入数据到ES（支持异步）
 * 4. 处理各种数据类型转换（数组、对象等）
 * 5. 实时追踪任务进度（内存缓存）
 * <p>
 * 安全特性：
 * 1. SQL注入防护：使用参数化查询
 * 2. 空指针保护：关键位置添加null检查
 * 3. 异常容错：完善的异常处理和日志记录
 * 4. 失败检测：批量导入时监控失败率，自动终止异常导入
 * <p>
 * 性能优化：
 * 1. 异步处理：大数据量导入不阻塞请求线程
 * 2. 批量操作：每批500条数据批量写入
 * 3. 进度追踪：实时更新任务进度，支持前端轮询查询
 * 4. 延迟清理：任务完成后延迟5秒清理缓存，确保前端获取最终状态
 *
 * @author lxw
 * @version V1.3
 * @date 2024-7-18
 * @updated 2025-10-31 新增异步处理、进度追踪、延迟清理功能
 */
@Service
public class EsProcessServiceImpl {

    private static final Logger log = LoggerFactory.getLogger(EsProcessServiceImpl.class);

    /**
     * Elasticsearch客户端
     * 用于执行ES操作（已测试可用）
     */
    private final ElasticsearchClient client;

    /**
     * 文件操作服务
     * 用于读取配置文件（已测试可用）
     */
    private final DynFileOperateService fileOperateService;

    /**
     * 进度缓存管理器
     * 负责任务进度的存储、查询和延迟清理
     */
    private final ProgressCacheManager progressCacheManager;


    /**
     * 构造函数 - 依赖注入
     *
     * @param client               Elasticsearch客户端（已测试可用）
     * @param fileOperateService   文件操作服务（已测试可用）
     * @param progressCacheManager 进度缓存管理器（已测试可用）
     */
    public EsProcessServiceImpl(ElasticsearchClient client,
                                DynFileOperateService fileOperateService,
                                ProgressCacheManager progressCacheManager) {
        this.client = client;
        this.fileOperateService = fileOperateService;
        this.progressCacheManager = progressCacheManager;
    }

    /**
     * 加载索引映射规则文件
     * 从项目资源中读取JSON格式的mapping配置文件
     * <p>
     * 使用场景：在创建ES索引前，需要先加载对应的映射规则
     *
     * @param mappingFileName mapping文件名称，带文件格式.json，如ruc_report.json
     * @return JSONObject 包含索引映射规则的JSON对象
     * @throws BusinessException 当文件读取失败或内容为空时抛出
     * @author lxw
     * @date 2025/3/20 14:04
     */
    public JSONObject loadMapping(String mappingFileName) {
        // 使用try-with-resources自动关闭流，防止资源泄露
        try (InputStream inputStream = fileOperateService.getInputStream(mappingFileName)) {
            // 读取输入流中的所有字节并转换为UTF-8字符串
            String mappingStr = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);

            // 验证读取的内容不为空
            if (StringUtils.isBlank(mappingStr)) {
                log.error("===> ES-Import mapping规则文件内容为空，文件名{}", mappingFileName);
                throw new BusinessException("ES-Import mapping规则文件内容为空！");
            }

            // 将JSON字符串解析为JSONObject对象（使用fastjson2库）
            JSONObject jsonObject = JSONObject.parseObject(mappingStr);
            Object properties = jsonObject.get("properties");
            if (properties == null || StringUtils.isBlank(properties.toString())) {
                log.error("===> ES-Import mapping规则文件properties为空，文件名{}", mappingFileName);
                throw new BusinessException("ES-Import mapping规则不完整！");
            }
            return jsonObject;
        } catch (Exception e) {
            // 记录错误日志，包含具体的异常信息
            log.error("===> ES-Import 读取mapping规则失败，文件名：{}，错误：{}", mappingFileName, e.getMessage());
            throw new BusinessException("ES-Import 无法获取文件内容，请检查文件路径或存储配置是否正确。");
        }
    }

    /**
     * 创建Elasticsearch索引结构
     * 如果索引已存在则跳过创建，直接返回true
     * <p>
     * 注意事项：
     * 1. 索引名称和别名不能相同
     * 2. 已存在的索引不会被删除，直接跳过创建流程
     * 3. 新创建索引使用1个主分片，0个副本分片的配置
     *
     * @param indexName  索引名称，如：user_info_20240101
     * @param indexAlias 索引别名，如：user_info
     * @param mapping    索引映射规则的JSON字符串
     * @return boolean true-创建成功或已存在，false-创建失败
     * @throws BusinessException 当索引名称与别名相同或创建失败时抛出
     * @author lxw
     * @date 2024-7-18
     */
    public boolean createEsIndex(String indexName, String indexAlias, String mapping) {

        try {
            // 校验：索引名称和别名不能相同，避免配置错误
            if (StringUtils.equalsIgnoreCase(indexName, indexAlias)) {
                log.error("===> ES-Import 生成ES索引结构，索引名称与索引别名不能相同");
                return false;
            }
            // 检查索引是否已存在（使用Elasticsearch Java客户端）
            BooleanResponse exists = client.indices().exists(e -> e.index(indexName));
            if (exists.value()) {
                // 如果索引已存在，跳过创建，直接返回true，继续执行数据导入
                log.info("===> ES-Import 生成ES索引结构，索引已存在，跳过创建！索引名称：{}", indexName);
                return true;
            }

            // 将mapping JSON字符串转换为InputStream，供ES客户端使用
            InputStream inputStream = new ByteArrayInputStream(mapping.getBytes(StandardCharsets.UTF_8));

            // 创建索引settings配置：1个主分片，0个副本分片
            // 使用工具类EsMappingUtil.initSetting初始化配置（已测试可用）
            IndexSettings.Builder settingBuilder = EsMappingUtil.initSetting(1, 0);

            // 构建创建索引请求
            CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder()
                    .index(indexName)                      // 设置索引名称
                    .aliases(indexAlias, a -> a)           // 设置索引别名
                    .settings(settingBuilder.build())      // 设置索引配置
                    .mappings(m -> m.withJson(inputStream)) // 设置字段映射规则
                    .build();

            // 执行创建索引操作
            CreateIndexResponse indexResponse = client.indices().create(createIndexRequest);
            boolean created = indexResponse.acknowledged();
            log.info("===> ES-Import 生成ES索引结构，索引创建响应: {}", created);
            return created;
        } catch (IOException e) {
            // IO异常处理：可能是网络问题或ES服务不可用
            log.error("===> ES-Import 生成ES索引结构，创建索引失败，索引名：{}，错误：{}", indexName, e.getMessage());
            return false;
        }
    }

    /**
     * 初始化ES索引数据（异步执行）
     * 完整流程：加载mapping规则 -> 创建索引 -> 导入数据
     * <p>
     * 这是一个组合方法，串联了索引创建和数据导入的完整流程
     * 使用@Async注解实现异步处理，避免大数据量导入时阻塞请求线程
     * <p>
     * 进度追踪：
     * 1. 任务开始时初始化进度缓存
     * 2. 导入过程中实时更新进度
     * 3. 任务完成后延迟清理缓存（5秒后），确保前端能获取最终状态
     * 4. 异常时立即清理缓存
     *
     * @param batchNo    批次号，用于标识和追踪任务进度
     * @param indexDTO   索引创建DTO，包含索引配置和数据源信息
     * @param dataSource 数据源对象，用于连接数据库
     * @author lxw
     * @date 2024-7-18
     * @updated 2025-10-31 新增异步处理、进度追踪、延迟清理功能
     */
    @Async("esImportExecutor")
    public void initEsData(String batchNo, CreateIndexDTO indexDTO, DataSource dataSource) {
        try {
            log.info("===> ES-Import 开始异步导入ES数据，批次号：{}", batchNo);
            // 第一步：读取索引映射规则文件
            JSONObject mappingJson = loadMapping(indexDTO.getRulePath());

            // 第二步：获取mapping规则中properties字段
            Object properties = mappingJson.get("properties");

            // 第三步：将properties转换为JSON字符串格式
            String jsonString = JSONObject.of("properties", properties).toString();

            // 第四步：创建ES索引结构
            // realIndexName()方法生成带时间戳的真实索引名
            if (!createEsIndex(indexDTO.realIndexName(), indexDTO.getIndexName(), jsonString)) {
                return;
            }
            // 第五步：导入数据到ES索引（传入批次号用于进度追踪）
            importEsData(batchNo, indexDTO, dataSource, mappingJson);

        } catch (Exception e) {
            // 发生异常时记录错误日志
            log.error("===> ES-Import 数据导入失败，批次号：{}，错误：{}", batchNo, e.getMessage(), e);
            // 异常时也延迟删除，确保前端能获取到异常信息
            // 标记任务为失败状态，传入异常消息
            progressCacheManager.markAsFailed(batchNo, 0L, 0L, "数据导入失败：" + e.getMessage());
        }
    }

    /**
     * 从数据库导入数据到Elasticsearch索引
     * 核心业务方法，处理完整的数据导入流程
     * <p>
     * 流程说明：
     * 1. 解析mapping配置，确定字段类型
     * 2. 查询数据总量，初始化进度缓存
     * 3. 分批次查询数据库数据（使用独立方法fetchData）
     * 4. 转换数据格式并批量写入ES
     * 5. 实时更新进度缓存
     * 6. 监控失败率，异常时自动终止
     * 7. 收集并记录失败的数据ID
     * 8. 任务完成后延迟清理缓存（5秒后）
     * <p>
     * 性能优化：
     * - 使用批量操作，每批500条数据
     * - 使用id游标分页，避免深分页性能问题
     * - 查询总量和批次查询使用独立方法，便于维护和测试
     * <p>
     * 安全特性：
     * - SQL注入防护：使用参数化查询
     * - 空指针保护：关键位置添加null检查
     * - 失败检测：失败率>50%时自动终止
     *
     * @param batchNo     批次号，用于追踪任务进度
     * @param indexDTO    索引创建DTO，包含数据源和表信息
     * @param dataSource  数据源对象
     * @param mappingJson 索引映射规则JSON对象
     * @throws BusinessException 当数据库连接不存在或规则为空时抛出
     * @author lxw
     * @date 2024-7-18
     * @updated 2025-10-31 新增进度追踪和延迟清理功能
     */
    public void importEsData(String batchNo, CreateIndexDTO indexDTO, DataSource dataSource, JSONObject mappingJson) {

        // ============ 第一部分：参数验证和初始化 ============
        // 验证数据源不能为空
        if (dataSource == null) {
            log.error("===> ES-Import 数据库源不存在！");
            // 异常时也延迟删除，确保前端能获取到异常信息
            // 标记任务为失败状态，传入异常消息
            progressCacheManager.markAsFailed(batchNo, 0L, 0L, "数据库源不存在!");
            return;
        }
        // ============ 第二部分：查询数据总量并初始化进度缓存 ============
        // 创建JdbcTemplate用于执行SQL查询
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);

        // 获取数据库表名和查询条件
        String tableName = indexDTO.getTableName();
        String condition = indexDTO.getCondition();

        // 调用独立方法查询总数据量（保持方法独立，不合并到此方法中）
        Long totalCount = getCount(jdbcTemplate, tableName, condition);
        log.info("===> ES-Import 批次号：{}，数据总量：{}", batchNo, totalCount);

        // 初始化进度缓存：总数totalCount，已完成0条
        progressCacheManager.updateProgress(batchNo, ImportProgressDTO.processing(totalCount, 0L));

        // 获取唯一ID字段名，用于ES文档ID
        String onlyId = Optional.ofNullable(mappingJson.get("request_id_field"))
                .map(Object::toString)
                .orElse("");

        // 解析特殊字段配置（数组、对象等）
        // 这些字段需要特殊处理，不能直接从数据库写入ES
        List<String> arrayFieldsAllowRepeat = processListString(mappingJson.get("array_field_allow_repeat"));
        List<String> arrayFieldsNotRepeat = processListString(mappingJson.get("array_field_not_repeat"));
        List<String> jsonArrayField = processListString(mappingJson.get("json_array_field"));
        List<String> jsonObjectField = processListString(mappingJson.get("json_object_field"));

        // 从mapping规则，获取properties配置，前置方法以判断，无需重复验证了
        Object properties = mappingJson.get("properties");
        // 提取所有字段名列表
        JSONObject from = JSONObject.from(properties);
        Set<String> keys = from.keySet();
        List<String> fieldList = new ArrayList<>(keys);

        // ============ 第三部分：批量查询和导入数据 ============
        // 初始化失败数据ID列表
        List<String> failedIds = new ArrayList<>();

        // 分批次查询和导入配置
        int cardinality = 500; // 每批次处理的数据量
//        int total = (int) Math.ceil(totalCount / (double) cardinality); // 总批次数
        String lastId = "0"; // 初始化游标ID，用于分页查询（从0开始）
        long processedCount = 0L; // 已处理的数据条数

        // 获取真实需要导入数据的ES索引名称
        String indexName = indexDTO.realIndexName();

        // 循环批量查询和导入数据
        while (true) {
            // 调用独立方法查询一批数据（保持方法独立，不合并到此方法中）
            // 使用id游标分页，避免深分页性能问题
            List<Map<String, Object>> dataList = fetchData(jdbcTemplate, tableName, condition, lastId, cardinality);
            // 如果没有数据了，退出循环
            if (dataList.isEmpty()) {
                break;
            }
            // 更新游标ID为本批次最后一条数据的ID
            // 添加null检查，防止空指针异常
            Object idObj = dataList.get(dataList.size() - 1).get("id");
            if (idObj == null) {
                log.error("===> ES-Import {}数据表，最后一条数据的id字段为null，终止导入！", tableName);
                // 游标ID异常为空，结束任务
                String errorMsg = "ES-Import 批量导入当前批次，最后一条数据的id字段为null，已终止导入";
                progressCacheManager.markAsFailed(batchNo, totalCount, processedCount, errorMsg);
                return;
            }
            lastId = idObj.toString();
            // 批量索引数据到ES
            BulkResponse bulkResponse = processBulkRequest(
                    indexName,
                    dataList,
                    onlyId,
                    fieldList,
                    arrayFieldsAllowRepeat,
                    arrayFieldsNotRepeat,
                    jsonArrayField,
                    jsonObjectField
            );

            // 处理批量操作结果
            if (bulkResponse != null) {
                // 检查是否有错误
                if (bulkResponse.errors()) {
                    handleBulkErrors(bulkResponse, failedIds);
                }

                // 计算本批次失败数量
                long failCount = bulkResponse.items().stream()
                        .filter(item -> item.error() != null)
                        .count();

                // 计算失败率
                double failRate = (double) failCount / dataList.size();

                // 如果失败率超过50%，终止导入
                if (failRate > 0.5) {
                    log.error("===> ES-Import 批次号：{}，批量导入失败率过高：{}%，终止导入！", batchNo, String.format("%.2f", failRate * 100));
                    // 失败率过高，延迟删除缓存并记录失败信息
                    String errorMsg = String.format("ES-Import 批量导入失败率过高：%.2f%%，已终止导入", failRate * 100);
                    progressCacheManager.markAsFailed(batchNo, totalCount, processedCount, errorMsg);
                    return;
                }
                log.info("===> ES-Import 批次号：{}，当前循环批量中，成功：{}，失败：{}", batchNo, dataList.size() - failCount, failCount);
            } else {
                // 批量操作返回null，说明请求失败
                log.error("===> ES-Import 批次号：{}，当前循环ES bulkRequest请求失败，已终止导入", batchNo);
                // 收集本批次所有数据的ID到失败列表
                failedIds.addAll(dataList.stream()
                        .map(m -> m.get(onlyId))
                        .filter(Objects::nonNull)
                        .map(Object::toString)
                        .toList());

                // 批量操作失败，延迟删除缓存并记录失败信息
                progressCacheManager.markAsFailed(batchNo, totalCount, processedCount, "批量请求失败，已终止导入");
                return;
            }

            // 更新已处理数量
            processedCount += dataList.size();

            // 更新进度缓存
            progressCacheManager.updateProgress(batchNo, ImportProgressDTO.processing(totalCount, processedCount));
            log.info("===> ES-Import 批次号：{}，进度更新：{}/{}", batchNo, processedCount, totalCount);
        }

        // ============ 第四部分：任务完成，延迟清理缓存 ============
        // 标记任务为完成状态，并启动延迟清理流程（5秒后删除缓存）
        // 这样前端轮询（每5秒一次）能够获取到最后一次进度结果
        // 使用实际的总量和完成量
        progressCacheManager.markAsCompleted(batchNo, totalCount, processedCount);
        log.info("===> ES-Import 批次号：{}，任务完成，总量：{}，完成量：{}，已启动延迟清理", batchNo, totalCount, processedCount);

        // 记录失败的数据ID列表
        if (!failedIds.isEmpty()) {
            log.warn("===> ES-Import 批次号：{}，导入完成，失败数量：{}，失败ID：{}", batchNo, failedIds.size(), failedIds);
        }
    }

    /**
     * 根据批次号获取任务进度
     * <p>
     * 使用场景：
     * - 前端页面轮询调用此方法，实时展示导入进度
     * - 任务完成后缓存会被延迟清理（5秒后），期间仍可查询到完成状态
     * - 缓存清理后返回100%完成状态
     *
     * @param batchNo 批次号
     * @return ImportProgressDTO 任务进度信息
     * @author lxw
     * @date 2025-10-31
     */
    public ImportProgressDTO getProgress(String batchNo) {
        // 委托给进度缓存管理器处理
        return progressCacheManager.getProgress(batchNo);
    }

    /**
     * 分页查询数据
     * 使用ID游标方式分页，避免OFFSET深分页性能问题
     * <p>
     * 技术要点：
     * - 使用 id > lastId 方式实现游标分页
     * - 按ID排序保证数据顺序和一致性
     * - 使用参数化查询防止SQL注入
     * <p>
     * 安全特性：
     * - lastId和cardinality使用参数化查询，安全
     * - condition直接拼接，需要调用方确保安全
     * <p>
     * 注意事项：
     * - 本方法保持独立，不合并到importEsData大方法中
     * - 便于单独测试和维护
     *
     * @param jdbcTemplate Spring JDBC模板对象（已测试可用）
     * @param tableName    表名（应来自配置或白名单）
     * @param condition    查询条件（WHERE子句内容）
     * @param lastId       上一批次最后一条数据的ID，用于游标分页
     * @param cardinality  每批次查询的数据条数
     * @return List<Map < String, Object>> 查询结果列表，每个Map代表一行数据
     * @author lxw
     * @date 2025/10/30 17:29
     * @updated 2025-10-31 增强异常处理和日志
     */
    private List<Map<String, Object>> fetchData(JdbcTemplate jdbcTemplate, String tableName,
                                                String condition, String lastId, int cardinality) {
        try {
            // 构建SQL语句
            StringBuilder sql = new StringBuilder("SELECT * FROM ");
            sql.append(tableName).append(" a WHERE a.id > ?");

            // 添加额外的查询条件
            // 注意：condition直接拼接，调用方需确保安全
            if (StringUtils.isNotBlank(condition)) {
                sql.append(" AND ").append(condition);
            }

            // 添加排序和分页限制
            sql.append(" ORDER BY a.id LIMIT ?");

            log.info("===> ES-Import 数据查询SQL: {}, 参数: lastId={}, limit={}", sql, lastId, cardinality);

            // 使用参数化查询，防止SQL注入
            // lastId和cardinality通过占位符传递，安全可靠
            return jdbcTemplate.queryForList(sql.toString(), lastId, cardinality);

        } catch (DataAccessException e) {
            // 捕获数据访问异常，记录详细日志后返回空列表
            log.error("===> ES-Import {}数据表查询数据错误，lastId: {}，错误: {}", tableName, lastId, e.getMessage());
            return new ArrayList<>();
        }
    }

    /**
     * 获取数据总条数
     * 根据表名和查询条件统计符合条件的数据量
     * <p>
     * 安全特性：
     * - 注意：tableName和condition直接拼接，调用方需确保这些参数来自可信源
     * - condition参数应该是预先验证过的安全条件
     * - 建议在调用方对condition进行白名单验证
     * <p>
     * 注意事项：
     * - 本方法保持独立，不合并到importEsData大方法中
     * - 便于单独测试和维护
     *
     * @param jdbcTemplate Spring JDBC模板对象（已测试可用）
     * @param tableName    表名（应来自配置或白名单）
     * @param condition    查询条件（WHERE子句内容，不包含WHERE关键字）
     * @return Long 符合条件的数据总数
     * @throws BusinessException 当数据量为0或查询失败时抛出
     * @author lxw
     * @date 2025/10/30 17:29
     * @updated 2025-10-31 增强安全性说明
     */
    private Long getCount(JdbcTemplate jdbcTemplate, String tableName, String condition) {
        // 构建基础SQL语句
        // 警告：tableName和condition直接拼接，必须确保来自可信源
        StringBuilder sql = new StringBuilder("SELECT COUNT(1) FROM ");
        sql.append(tableName);

        // 如果有查询条件，添加WHERE子句
        if (StringUtils.isNotBlank(condition)) {
            sql.append(" WHERE ").append(condition);
        }

        try {
            // 执行查询获取数据总数
            Long count = jdbcTemplate.queryForObject(sql.toString(), Long.class);

            // 验证查询结果,处理count为null的情况，防止空指针异常
            if (count == null || count <= 0L) {
                throw new BusinessException("本次无数据需要收割。");
            }

            log.info("===> ES-Import 数据表总量查询，表：{}，总记录数：{}", tableName, count);
            return count;
        } catch (DataAccessException e) {
            log.error("===> ES-Import 数据表总量查询，表：{}，查询总数失败：{}", tableName, e.getMessage());
            throw new BusinessException("ES-Import 查询数据总数失败！");
        }
    }

    /**
     * 创建mapping的JSON字符串
     * 将properties对象包装成完整的mapping结构
     * <p>
     * ES mapping格式：
     * {
     * "properties": {
     * "field1": { "type": "text" },
     * "field2": { "type": "keyword" }
     * }
     * }
     *
     * @param properties 字段映射配置对象
     * @return String JSON格式的mapping字符串
     * @author lxw
     * @date 2024/12/9 9:16
     */
    private String createJsonString(Object properties) {
        // 创建包装Map，将properties放入其中
        Map<String, Object> mm = new HashMap<>();
        mm.put("properties", properties);


        // 使用fastjson2序列化为JSON字符串
        return JSONObject.toJSONString(mm);
    }

    /**
     * 批量将数据写入Elasticsearch
     * 核心数据转换和索引方法
     * <p>
     * 处理流程：
     * 1. 遍历原始数据列表
     * 2. 根据字段配置转换数据格式（调用processArrayFields方法）
     * 3. 构建批量索引请求
     * 4. 执行批量写入操作
     * <p>
     * 性能优化：
     * - 使用BulkRequest批量操作
     * - 设置Refresh.True立即刷新索引
     *
     * @param indexName              索引名称
     * @param list                   原始数据列表
     * @param onlyId                 原数据中唯一值字段名，用作ES文档ID
     * @param fieldList              ES索引需要的字段列表
     * @param arrayFieldsAllowRepeat ES索引数据值为字符串数组，且允许值重复的字段
     * @param arrayFieldsNotRepeat   ES索引数据值为字符串数组，且不允许值重复的字段
     * @param jsonArrayField         ES索引数据值为数组对象的字段（对象数组）
     * @param jsonObjectField        ES索引数据值为对象的字段（单个对象）
     * @return BulkResponse ES批量操作响应对象，包含成功和失败信息
     * @author lxw
     * @date 2024/12/9 9:19
     */
    private BulkResponse processBulkRequest(String indexName, List<Map<String, Object>> list,
                                            String onlyId, List<String> fieldList,
                                            List<String> arrayFieldsAllowRepeat,
                                            List<String> arrayFieldsNotRepeat,
                                            List<String> jsonArrayField,
                                            List<String> jsonObjectField) {
        // 创建批量请求构建器
        BulkRequest.Builder bulkRequest = new BulkRequest.Builder();

        // 遍历每条数据，转换并添加到批量请求中
        for (Map<String, Object> map : list) {
            // 处理特殊字段：数组字段、对象字段等
            // processArrayFields方法会根据配置转换数据格式
            JSONObject jsonObject = processArrayFields(
                    map,
                    fieldList,
                    arrayFieldsAllowRepeat,
                    arrayFieldsNotRepeat,
                    jsonArrayField,
                    jsonObjectField
            );

            // 获取文档ID
            // 增强onlyId空字符串判断
            String id;
            if (StringUtils.isNotBlank(onlyId)) {
                // onlyId不为空，从map中获取对应字段值
                Object idObj = map.get(onlyId);
                id = (idObj != null) ? idObj.toString() : UUIDUtils.getInstance().generateShortUuid();
            } else {
                // onlyId为空，直接生成UUID
                id = UUIDUtils.getInstance().generateShortUuid();
            }

            // 添加索引操作到批量请求
            bulkRequest.operations(op -> op
                    .index(idx -> idx
                            .index(indexName)      // 指定索引名称
                            .id(id)                // 指定文档ID
                            .document(jsonObject)  // 设置文档内容
                    )
            );
        }

        // 设置立即刷新，使数据立即可搜索
        bulkRequest.refresh(Refresh.True);

        try {
            // 执行批量索引操作
            return client.bulk(bulkRequest.build());
        } catch (IOException e) {
            // 记录错误日志
            log.error("===> ES-Import {}索引批量创建失败，ES错误: {}", indexName, e.getMessage());
            return null;
        }
    }

    /**
     * 数据格式转换为ES索引需要的格式
     * 根据字段配置转换不同类型的数据
     * <p>
     * 支持的数据类型转换：
     * 1. 字符串数组（允许重复）："a||b||a" -> ["a", "b", "a"]
     * 2. 字符串数组（不允许重复）："a||b||a" -> ["a", "b"]
     * 3. JSON对象数组：'[{"id":1},{"id":2}]' -> JSONArray
     * 4. JSON对象：'{"name":"test"}' -> JSONObject
     *
     * @param map                    原始数据Map
     * @param fieldList              ES索引字段集合
     * @param arrayFieldsAllowRepeat ES索引数据值为字符串数组，且允许值重复的字段
     * @param arrayFieldsNotRepeat   ES索引数据值为字符串数组，且不允许值重复的字段
     * @param jsonArrayField         ES索引数据值为数组对象的字段（对象数组）
     * @param jsonObjectField        ES索引数据值为对象的字段（单个对象）
     * @return JSONObject 转换后的ES文档对象
     * @author lxw
     * @date 2024/12/9 9:20
     */
    private JSONObject processArrayFields(Map<String, Object> map, List<String> fieldList,
                                          List<String> arrayFieldsAllowRepeat, List<String> arrayFieldsNotRepeat,
                                          List<String> jsonArrayField, List<String> jsonObjectField) {

        // 处理允许重复的字符串数组字段
        // 例如：tags字段 "Java||Python||Java" -> ["Java", "Python", "Java"]
        arrayFieldsAllowRepeat.forEach(field ->
                map.put(field, processArrayString(map.get(field), false))
        );

        // 处理不允许重复的字符串数组字段
        // 例如：category字段 "A||B||A" -> ["A", "B"]
        arrayFieldsNotRepeat.forEach(field ->
                map.put(field, processArrayString(map.get(field), true))
        );

        // 处理JSON数组对象字段
        // 例如：items字段 '[{"id":1,"name":"item1"}]' -> JSONArray对象
        jsonArrayField.forEach(field ->
                map.put(field, processArrayObject(map.get(field)))
        );

        // 处理JSON对象字段
        // 例如：userInfo字段 '{"name":"张三","age":20}' -> JSONObject对象
        jsonObjectField.forEach(field ->
                map.put(field, processObject(map.get(field)))
        );

        // 创建最终的ES文档对象，只包含fieldList中指定的字段
        JSONObject jsonObject = new JSONObject();
        fieldList.forEach(key -> {
            Object object = map.get(key);
            if (object == null) {
                jsonObject.put(key, "");
            } else {
                jsonObject.put(key, object);
            }
        });

        return jsonObject;
    }

    /**
     * 字符串按 || 分割转换为数组
     * 将数据库中的分隔符字符串转换为ES的数组格式
     * <p>
     * 使用场景：
     * - 数据库字段：tags = "Java||Python||MySQL"
     * - 转换后ES字段：tags = ["Java", "Python", "MySQL"]
     *
     * @param fieldValue 字段值，格式如："value1||value2||value3"
     * @param distinct   是否去重（true-不允许重复，false-允许重复）
     * @return List<String> 字符串数组，如果fieldValue为null则返回null
     * @author lxw
     * @date 2024/12/9 9:21
     */
    private List<String> processArrayString(Object fieldValue, boolean distinct) {
        // 空值处理
        if (fieldValue == null) return null;

        // 根据distinct参数选择不同的转换方法
        // EsStrUtil工具类提供字符串分割和去重功能（已测试可用）
        return distinct
                ? EsStrUtil.str2ListNotRepeat(fieldValue.toString())    // 去重
                : EsStrUtil.str2ListAllowRepeat(fieldValue.toString()); // 允许重复
    }

    /**
     * Object转换为JSONObject
     * 将数据库中的JSON字符串转换为fastjson2的JSONObject对象
     * <p>
     * 使用场景：
     * - 数据库字段：user_info = '{"name":"张三","age":20}'
     * - 转换后ES字段：user_info = JSONObject对象
     *
     * @param fieldValue 字段值，应为JSON格式的字符串
     * @return JSONObject JSON对象，如果转换失败或值为null则返回null
     * @author lxw
     * @date 2024/12/9 10:43
     * @updated 2025-10-31 增强异常日志，包含原始数据
     */
    private JSONObject processObject(Object fieldValue) {
        if (fieldValue != null) {
            try {
                // 使用fastjson2解析JSON字符串为JSONObject
                return JSONObject.parseObject(fieldValue.toString());
            } catch (Exception e) {
                // 增强错误日志，输出导致解析失败的原始数据
                log.error("===> ES-Import Object 转 JSONObject 失败！原始数据：{}，错误：{}", fieldValue, e.getMessage());
            }
        }
        return new JSONObject();
    }

    /**
     * Object转换为JSONArray
     * 将数据库中的JSON数组字符串转换为fastjson2的JSONArray对象
     * <p>
     * 使用场景：
     * - 数据库字段：order_items = '[{"id":1,"price":100},{"id":2,"price":200}]'
     * - 转换后ES字段：order_items = JSONArray对象
     *
     * @param fieldValue 字段值，应为JSON数组格式的字符串
     * @return JSONArray JSON数组对象，如果转换失败或值为null则返回null
     * @author lxw
     * @date 2024/12/9 10:43
     * @updated 2025-10-31 增强异常日志，包含原始数据
     */
    private JSONArray processArrayObject(Object fieldValue) {
        if (fieldValue != null) {
            try {
                // 使用fastjson2解析JSON数组字符串为JSONArray
                return JSONArray.parseArray(fieldValue.toString());
            } catch (Exception e) {
                // 增强错误日志，输出导致解析失败的原始数据
                log.error("===> ES-Import Object 转 JSONArray 失败！原始数据：{}，错误：{}", fieldValue, e.getMessage());
            }
        }
        return new JSONArray();
    }

    /**
     * Object转换为List<String>
     * 将配置文件中的字符串数组转换为Java的List集合
     * <p>
     * 使用场景：
     * - 配置文件：array_field_allow_repeat = '["tags","keywords"]'
     * - 转换后：List<String> = ["tags", "keywords"]
     *
     * @param fieldValue 字段值，应为JSON数组格式的字符串
     * @return List<String> 字符串列表，如果值为null则返回空列表
     * @author lxw
     * @date 2024/12/9 10:45
     */
    private static List<String> processListString(Object fieldValue) {
        // 使用fastjson2解析JSON数组为List<String>
        // 如果fieldValue为null，返回空列表而不是null，避免后续NPE
        return (fieldValue == null)
                ? Collections.emptyList()
                : JSONArray.parseArray(fieldValue.toString(), String.class);
    }


    /**
     * 处理批量索引操作的错误结果
     * 从BulkResponse中提取失败的文档ID
     * <p>
     * 使用场景：
     * - 批量导入数据后，需要知道哪些数据导入失败
     * - 后续可以针对失败的数据进行重试或人工处理
     *
     * @param bulkResponse ES批量操作响应对象（已测试可用）
     * @param failedIds    失败数据ID集合（传入的列表会被修改）
     * @author lxw
     * @date 2024/12/9 9:25
     */
    private void handleBulkErrors(BulkResponse bulkResponse, List<String> failedIds) {
        // 使用Stream API过滤出有错误的条目，并添加到失败列表
        bulkResponse.items().stream()
                .filter(item -> item.error() != null)  // 只处理有错误的条目
                .forEach(item -> {
                    // 将错误的文档ID添加到失败列表
                    failedIds.add(item.id());
                    // 记录详细的错误信息，便于排查问题
                    log.warn("===> ES-Import 文档索引失败，ID：{}，错误：{}", item.id(), item.error().reason());
                });
    }

}